package com.etone.esper.main.epl_sql;

import com.espertech.esper.client.Configuration;
import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPServiceProviderManager;
import com.espertech.esper.client.EPStatement;
import com.etone.esper.main.StatmentUpdater1;

public class ESPer {

	public static long sentEventT = 0l;
	
	public long eventSize = 0;
	public long limitSize = 0;
	
	public long timeWindow = 0;
	public long sleepTime = 0;
	
	private EPServiceProvider ESPEngine;
	
	public ESPer(long eventSize, long limitSize, long timeWindow, long sleepTime){
		this.eventSize = eventSize;
		this.limitSize = limitSize;
		this.timeWindow = timeWindow;
		init();
	}
	
	private void init(){
		Configuration configuration = new Configuration();
		configuration.configure("com/etone/esper/main/epl_sql/esper.cfg.xml");
		
		
		ESPEngine = EPServiceProviderManager.getDefaultProvider(configuration);
		
		addEPStatment();
	}

	public void addEPStatment(){
	
//		String stmtDB = "select irstream id ,age,count from Event_1, sql:topo['select age,count(1) as count from _esper limit 10,10']";
		String stmtDB = "select istream id ,kid,age from Event_1.win:time_batch(" + timeWindow + " sec), sql:topo['select id as kid,age from _esper where id=${id}']";
		String stmtDB1 = "select * from Event_1.win:time_batch(1 sec)";
		EPStatement epstmtDB = ESPEngine.getEPAdministrator().createEPL(stmtDB);
		EPStatement epstmtDB1 = ESPEngine.getEPAdministrator().createEPL(stmtDB1);
		
//		epstmtDB.addListener(new Updater());
		epstmtDB1.addListener(new StatmentUpdater1());
		
	}

	public void setEvent(){
	
		while(true){
			try {
				long start = System.currentTimeMillis();
				for(int i=1;i<=eventSize;i++){
					Event_1 e1 = new Event_1();
					e1.setId(i);
					e1.setName("jasic");
					
					ESPEngine.getEPRuntime().sendEvent(e1);
				}
				
				long end = System.currentTimeMillis();
				long costTime = (end - start);
				System.out.println("send " + eventSize + " event cost: " +  costTime + "ms");
				Thread.sleep(timeWindow * 1000 - costTime);
				sentEventT = System.currentTimeMillis();
				
				Thread.sleep(sleepTime * 1000);
				Thread.sleep(999999999);
				
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		
		
	}
	
	public static void main(String[] args) {
		int a1 = 1;
		int a2 = 100;
		int a3 = 10;
		int sleepTime = 10;
		
		if(args != null && args.length != 0){
			a1 = Integer.parseInt(args[0]);
			a2 = Integer.parseInt(args[1]);
			a3 = Integer.parseInt(args[2]);
		}
		
		new ESPer(a1,a2,a3,sleepTime).setEvent();
	}
}
